package org.atguigu.gmall.realtime.util;

import org.atguigu.gmall.realtime.common.Constant;

/**
 * @Author zenghaifeng
 * @Date 2023/4/24 21:24
 * @Package: org.atguigu.gmall.realtime.util
 * @Class: SQLUtil
 * @Description: TODO
 * @Version 1.0
 */
public class SQLUtil {
    /**
     * @Author: zenghaifeng
     * @Date: 2023/4/24 21:36
     * @TODO: TODO build a kafka source DDL .
     * @param: topic
     * @param: groupId
     * @param: format
     * @return: java.lang.String
     */
    public static String getKafkaSourceDDL(String topic,String groupId,String... format){
        String defaultFormat = "json";
        if (format.length > 0){
            defaultFormat = format[0];
        }
        return "with(" +
                " 'connector' = 'kafka', " +
                " 'topic' = '"+topic+"', " +
                " 'properties.bootstrap.servers' = '"+ Constant.KAFKA_SERVERS +"', " +
                " 'properties.group.id' = '"+groupId+"', " +
                " 'scan.startup.mode' = 'latest-offset', " +
                " 'format' = '"+defaultFormat+"' " +
                ")";
    }

    public static String getKafkaSinkDDL(String topic, String... format){
        String defaultFormat = "json";
        if (format.length > 0) {
            defaultFormat = format[0];
        }
        return "with(" +
                "  'connector' = 'kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '" + Constant.KAFKA_SERVERS + "', " +
                ("json".equals(defaultFormat) ? " 'json.ignore-parse-errors' = 'true', " : "") +
                "  'format' = '" + defaultFormat + "' " +
                ")";
    }

    public static String getUpsertKafkaDDL(String topic, String ... format) {
        String defaultFormat = "json";
        if (format.length > 0) {
            defaultFormat = format[0];
        }
        return "with(" +
                "  'connector' = 'upsert-kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '" + Constant.KAFKA_SERVERS + "', " +
                ("json".equals(defaultFormat) ? " 'key.json.ignore-parse-errors' = 'true', " : "") +
                ("json".equals(defaultFormat) ? " 'value.json.ignore-parse-errors' = 'true', " : "") +
                "  'key.format' = '" + defaultFormat + "',  " +
                "  'value.format' = '" + defaultFormat + "'  " +
                ")";
    }

}
